package com.tod.spark.springbootspark.sql;

import com.tod.spark.springbootspark.sql.model.Student;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.util.List;
import java.util.regex.Pattern;

/**
 * RDD convert to DataFrame
 * @author Tod.xie
 */
public class RDD2DataFrameReflection {
    private static final Pattern SPACE = Pattern.compile(",");

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("DateFrame").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(jsc);

        JavaRDD<String> lines = jsc.textFile("src/main/resources/static/student.txt");
        JavaRDD<Student> students = lines.map(line -> {
            String[] lineSplit = SPACE.split(line);
            return new Student(Integer.valueOf(lineSplit[0]), lineSplit[1], Integer.valueOf(lineSplit[2]));
        });

        //use reflect Type ,RDD convert to DataFrame
        Dataset<Row> studentDF = sqlContext.createDataFrame(students, Student.class);
        System.out.println("***************************************************************");
        studentDF.printSchema();
        System.out.println("***************************************************************");
        // Have a DataFrame,register to a temp table,Query age<= 18 young people by sql
        studentDF.registerTempTable("students");
        Dataset<Row> teenagerDF = sqlContext.sql("select * from students where age <= 18");

        //DataFrame convert to RDD
        JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD();
        System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$");
        JavaRDD<Student> studentRDD = teenagerRDD.map((Function<Row, Student>) row -> {
            //Row的顺序这里有可能被大乱
            System.out.println(row.get(0));
            System.out.println(row.get(1));
            System.out.println(row.get(2));
            return new Student(row.getInt(1), row.getString(2), row.getInt(0));
        });
        System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$");

        List<Student> studentsResult = studentRDD.collect();
        System.out.println("%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%");
        studentsResult.forEach(student -> System.out.println(student));
        System.out.println("%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%");

        jsc.stop();
    }
}
